
/**
 * Date:2021年9月22日下午8:14:16
 * Copyright (c) 2021, www.bwbroad.com All Rights Reserved.
 *
*/

package com.example.demo;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import com.alibaba.fastjson.JSON;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
import com.github.shyiko.mysql.binlog.event.Event;
import com.github.shyiko.mysql.binlog.event.EventData;
import com.github.shyiko.mysql.binlog.event.EventHeader;
import com.github.shyiko.mysql.binlog.event.EventHeaderV4;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.RotateEventData;
import com.github.shyiko.mysql.binlog.event.TableMapEventData;
import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData;
import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;

/**
 * Description: TODO <br/>
 * Date:     2021年9月22日 下午8:14:16 <br/>
 * @author   wangbing	 
 */

public class Snippet {
	    private static final SyncConfig syncConfig;

	    private static IPositionHandler positionHandler = new DefaultPositionHandler();

	    /**
	     * 项目启动读取的配置
	     */
	    static {
	        syncConfig = new SyncConfig();

	        syncConfig.setHost("127.0.0.1");
	        syncConfig.setPort(3307);
	        syncConfig.setUserName("root");
	        syncConfig.setPassword("123");
	    }

	    public static void main(String[] args) throws IOException {

	        BinaryLogClient client = new BinaryLogClient(syncConfig.getHost(), syncConfig.getPort(), syncConfig.getUserName(), syncConfig.getPassword());
	        EventDeserializer eventDeserializer = new EventDeserializer();
	        client.setEventDeserializer(eventDeserializer);
	        client.setServerId(8023);
	        //获取position的位置（创建client时，读取当前记录的postion）
	        BinlogPositionEntity binlogPositionEntity = positionHandler.getPosition(syncConfig);
	        if (binlogPositionEntity != null &&
	                binlogPositionEntity.getBinlogName() != null &&
	                binlogPositionEntity.getPosition() != null) {
	            client.setBinlogFilename(binlogPositionEntity.getBinlogName());
	            client.setBinlogPosition(binlogPositionEntity.getPosition());
	        }

	        client.registerEventListener(new BinaryLogClient.EventListener() {

	            @Override
	            public void onEvent(Event event) {

	                EventHeader header = event.getHeader();

	                EventType eventType = header.getEventType();
	                System.out.println("监听的事件类型:" + eventType);

	                /*
	                 * 不计入position更新的事件类型
	                 * FORMAT_DESCRIPTION类型为binlog起始时间
	                 * HEARTBEAT为心跳检测事件，不会写入master的binlog，记录该事件的position会导致重启时报错
	                 */
	                List<EventType> excludePositionEventType = new ArrayList<>();
	                excludePositionEventType.add(EventType.FORMAT_DESCRIPTION);
	                excludePositionEventType.add(EventType.HEARTBEAT);
	                if (!excludePositionEventType.contains(eventType)) {
	                    BinlogPositionEntity binlogPositionEntity = new BinlogPositionEntity();
	                    //处理rotate事件,这里会替换调binlog fileName
	                    if (event.getHeader().getEventType().equals(EventType.ROTATE)) {
	                        RotateEventData rotateEventData = (RotateEventData) event.getData();
	                        System.out.println(rotateEventData);
	                        binlogPositionEntity.setBinlogName(rotateEventData.getBinlogFilename());
	                        binlogPositionEntity.setPosition(rotateEventData.getBinlogPosition());
	                        binlogPositionEntity.setServerId(event.getHeader().getServerId());
	                    } else { //统一处理事件对应的binlog position
	                        //在Redis中获取获取binlog的position配置
	                        binlogPositionEntity = positionHandler.getPosition(syncConfig);
	                        EventHeaderV4 eventHeaderV4 = (EventHeaderV4) event.getHeader();
	                        binlogPositionEntity.setPosition(eventHeaderV4.getPosition());
	                        binlogPositionEntity.setServerId(event.getHeader().getServerId());
	                    }
	                    //将最新的配置保存到Redis中
	                    positionHandler.savePosition(syncConfig, binlogPositionEntity);

	                    //todo 解析结果
	                    EventData data = event.getData();
	        	        System.out.println(event);
	        	        System.out.println(data);
	        	        if (data instanceof TableMapEventData) {
	        	            System.out.println("Table:");
	        	            TableMapEventData tableMapEventData = (TableMapEventData) data;
	        	            System.out.println(tableMapEventData.getTableId()+": ["+tableMapEventData.getDatabase() + "-" + tableMapEventData.getTable()+"]");
	        	        }
	        	        if (data instanceof UpdateRowsEventData) {
	        	            System.out.println("Update:");
	        	            System.out.println(data.toString());
	        	        } else if (data instanceof WriteRowsEventData) {
	        	            System.out.println("Insert:");
	        	            System.out.println(data.toString());
	        	        } else if (data instanceof DeleteRowsEventData) {
	        	            System.out.println("Delete:");
	        	            System.out.println(data.toString());
	        	        }

	                }


	            }
	        });
	        client.connect();
	    }

}

